package org.senseframework.support.orm.hbase.query;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnPaginationFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.senseframework.support.orm.hbase.annotation.ColumnType;
import org.senseframework.support.orm.hbase.datasource.HBaseDataSource;
import org.senseframework.support.orm.hbase.definition.HBaseColumnDefinition;
import org.senseframework.support.orm.hbase.definition.HBaseEntityDefinition;
import org.senseframework.support.orm.hbase.utils.HBaseEntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author wangqian
 * @date 18/4/11 下午2:53
 */
public class HBase extends HBaseQuery {

    private static final Logger logger = LoggerFactory.getLogger(HBase.class);

    private HbaseTemplate hbaseTemplate;
    private HBaseEntityDefinition entityDefinition;

    private HBase(Class clazz) {
        super(clazz);
        this.hbaseTemplate = HBaseDataSource.getSource(clazz);
        this.entityDefinition = HBaseEntityUtils.findEntityDefinition(getClazz());
    }

    public static HBase build(Class clazz) {
        return new HBase(clazz);
    }

    public HBase row(String rowKey) {
        setRowKey(rowKey);
        return this;
    }

    public HBase row(List<String> rowKeys) {
        setRowKeys(rowKeys);
        return this;
    }

    public HBase startRow(String startRow) {
        scan.setStartRow(Bytes.toBytes(startRow));
        return this;
    }

    public HBase endRow(String endRow) {
        scan.setStopRow(Bytes.toBytes(endRow));
        return this;
    }

    public HBase columnPageFilter(int limit, int page) {
        int offset = (page - 1) * limit;
        scan.setFilter(new ColumnPaginationFilter(limit, offset));
        return this;
    }

    public HBase size(Integer pageSize) {
        setSize(pageSize);
        return this;
    }

    private HBase add(String family, String qualifier) {
        getScan().addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        return this;
    }

    private List<Update> updates = null;

    public <T> List<T> list() {
        if (CollectionUtils.isNotEmpty(rowKeys)) {
            return hbaseTemplate.execute(tableName(), new TableCallback<List<T>>() {
                @Override
                public List<T> doInTable(HTableInterface hTableInterface) throws Throwable {
                    List<Get> gets = new ArrayList<Get>();
                    for (String rowKey : getRowKeys()) {
                        Get get = new Get(Bytes.toBytes(rowKey));
                        gets.add(get);
                    }
                    Result[] results = hTableInterface.get(gets);
                    return entityDefinition.resultToEntity(results, getClazz());
                }
            });
        } else {
            if (scan.getStartRow() == null || scan.getStartRow().length == 0) {
                logger.error("start row is empty");
                return null;
            }
            List<Filter> filterList = new ArrayList<Filter>();
            if (getSize() > 0) {
                filterList.add(new PageFilter(getSize()));
            }
            if (filterList.size() > 0) {
                FilterList f = new FilterList(filterList.toArray(new Filter[filterList.size()]));
                scan.setFilter(f);
            }
            return hbaseTemplate.find(tableName(), scan, new RowMapper<T>() {
                @Override
                public T mapRow(Result result, int i) throws Exception {
                    return (T) entityDefinition.resultToEntity(result, getClazz());
                }
            });
        }
    }

    public <T> T one() {
        if (scan != null && scan.getFilter() instanceof ColumnPaginationFilter) {
            scan.setStartRow(Bytes.toBytes(getRowKey())).setStopRow(Bytes.toBytes(getRowKey()));
            //扫描过滤
            return hbaseTemplate.execute(tableName(), new TableCallback<T>() {
                @Override
                public T doInTable(HTableInterface hTableInterface) throws Throwable {
                    ResultScanner scanner = hTableInterface.getScanner(scan);
                    Iterator i = scanner.iterator();
                    while (i.hasNext()) {
                        Result result = (Result) i.next();
                        return result == null ? null : (T) entityDefinition.resultToEntity(result, getClazz());
                    }
                    return null;
                }
            });
        } else {
            //查找单个数据
            return hbaseTemplate.get(tableName(), getRowKey(), new RowMapper<T>() {
                @Override
                public T mapRow(Result result, int i) throws Exception {
                    return (T) entityDefinition.resultToEntity(result, getClazz());
                }
            });
        }
    }

    public <T> T get(String family, String qualifier) {
        return hbaseTemplate.get(tableName(), getRowKey(), family, qualifier, new RowMapper<T>() {
            @Override
            public T mapRow(Result result, int i) throws Exception {
                return (T) entityDefinition.resultToEntity(result, getClazz());
            }
        });
    }

    public <T> T incr(final String family, final String qualifier, final long amount) {
        return hbaseTemplate.execute(tableName(), new TableCallback<T>() {
            @Override
            public T doInTable(HTableInterface hTableInterface) throws Throwable {
                Increment increment = new Increment(Bytes.toBytes(getRowKey()));
                increment.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), amount);
                Result result = hTableInterface.increment(increment);
                return (T) entityDefinition.resultToEntity(result, getClazz());
            }
        });
    }

    public HBase set(String family, Object value) {
        addSet(family, null, value);
        return this;
    }

    public HBase set(String family, Map<String, Object> qualifierValueMap) {
        for (String qualifier : qualifierValueMap.keySet()) {
            Object data = qualifierValueMap.get(qualifier);
            addSet(family, qualifier, data);
        }
        return this;
    }

    public HBase set(String family, String qualifier, Object value) {
        addSet(family, qualifier, value);
        return this;
    }

    public HBase set(Object object) {
        if (!getClazz().equals(object.getClass())) {
            throw new RuntimeException("不支持该类");
        }
        if (MapUtils.isNotEmpty(entityDefinition.getColumnDefinitions())) {
            for (HBaseColumnDefinition columnDefinition : entityDefinition.getColumnDefinitions().values()) {

                try {
                    if (columnDefinition.getColumnType().equals(ColumnType.QUALIFIER)) {
                        Map<String, Object> map = (Map) PropertyUtils.getProperty(object, columnDefinition.getJavaColumn());
                        if (MapUtils.isNotEmpty(map)) {
                            for (String key : map.keySet()) {
                                addSet(columnDefinition.getFamily(), key, map.get(key));
                            }
                        }
                    } else {
                        Object value = PropertyUtils.getProperty(object, columnDefinition.getJavaColumn());
                        if (value != null) {
                            addSet(columnDefinition.getFamily(), columnDefinition.getQualifier(), value);
                        }
                    }
                } catch (Exception e) {
                    logger.error("set object", e);
                }

            }
        }
        try {
            Object id = PropertyUtils.getProperty(object, entityDefinition.getIdJavaColumn());
            if (id != null) {
                setRowKey(String.valueOf(id));
            }
        } catch (Exception e) {
            logger.error("set object", e);
        }
        return this;
    }

    private void addSet(String family, String qualifier, Object value) {
        synchronized (this) {
            if (updates == null) {
                updates = new ArrayList<Update>();
            }
            Update update = new Update(family, qualifier, entityDefinition.getSerialize().encode(value));
            updates.add(update);
        }
    }

    public void update() {
        save();
    }

    public <T> void save() {
        try {
            hbaseTemplate.execute(tableName(), new TableCallback<T>() {
                @Override
                public T doInTable(HTableInterface table) throws Throwable {
                    Put put = new Put(Bytes.toBytes(getRowKey()));
                    for (Update update : updates) {
                        put.add(update.getFamily(), update.getQualifier(), update.getValue());
                    }
                    table.put(put);
                    return null;
                }
            });
        } catch (Exception e) {
            logger.error("save", e);
        }
    }

    public void delete() {
        hbaseTemplate.execute(tableName(), new TableCallback<Object>() {
            @Override
            public Object doInTable(HTableInterface htable) throws Throwable {
                Delete delete = new Delete(Bytes.toBytes(getRowKey()));
                htable.delete(delete);
                return null;
            }
        });
    }

    public void delete(String family) {
        hbaseTemplate.delete(tableName(), getRowKey(), family);
    }

    public void delete(String family, String qualifier) {
        hbaseTemplate.delete(tableName(), getRowKey(), family, qualifier);
    }

    private String tableName() {
        return entityDefinition.getTable();
    }
}
